package org.sn.jdish.spark.sql;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.SQLContext;

/**
 * DataFrame的操作
 * 
 * @author snzigod@hotmail.com
 */
public class DataFrameOperation {

	public static void main(String[] args) {
		/**
		 * SparkConf:第一步创建一个SparkConf，在这个对象里面可以设置允许模式Local Standalone yarn
		 * AppName(可以在Web UI中看到) 还可以设置Spark运行时的资源要求
		 */
		SparkConf sparkConf = new SparkConf().setAppName("sparkSql").setMaster("local");

		/**
		 * 基于SparkConf的对象可以创建出来一个SparkContext Spark上下文
		 * SparkContext是通往集群的唯一通道，SparkContext在创建的时候还会创建任务调度器
		 */
		JavaSparkContext javaSparkContext = new JavaSparkContext(sparkConf);

		readJson(javaSparkContext);

		javaSparkContext.close();
	}

	public static void readJson(JavaSparkContext javaSparkContext) {
		SQLContext sqlContext = new SQLContext(javaSparkContext);

		// 将数据源读取为数据框,可以理解为一张表。具有数据和结构信息
		DataFrame df = sqlContext.read().json("data/people.json");

		// 格式化的打印这张表
		df.show();

		// 搭建元数据(结构)schema
		df.printSchema();

		// 查询列并计算
		df.select("name").show();
		df.select(df.col("name"), df.col("age").plus(1)).show();

		// 过滤
		df.filter(df.col("age").gt(20)).show();

		// 根据某一列分组然后统计count
		df.groupBy("age").count().show();
	}
}
